#include <stdio.h>
#include <stdlib.h>
#include <string.h>

#include "ipcapi.h"

#if !defined(IPCAPI_USE_ZMQ) && !defined(IPCAPI_USE_NANOMSG)
#define IPCAPI_USE_NANOMSG
#endif

#define IPCAPI_INFO(...) fprintf(stdout, "[IPCAPI_INFO] %s(%d): ", __FUNCTION__, __LINE__),fprintf(stdout, __VA_ARGS__)
#define IPCAPI_ERR(...) fprintf(stderr, "[IPCAPI_ERR] %s(%d): ", __FUNCTION__, __LINE__),fprintf(stderr, __VA_ARGS__)

#if defined(WIN32) || defined(WIN64) || defined(WINCE) || defined(_MSC_VER)
/* ----- windows begin ----- */
#include <windows.h>
#include <process.h> /* _beginthread */
#if defined(IPCAPI_USE_ZMQ)
#pragma comment(lib, "ws2_32.lib")
#pragma comment(lib, "libzmq.lib")
#elif defined(IPCAPI_USE_NANOMSG)
#pragma comment(lib, "ws2_32.lib")
#pragma comment(lib, "mswsock.lib")
#pragma comment(lib, "advapi32.lib")
#pragma comment(lib, "nanomsg.lib")
#endif

#define IPCAPI_SLEEP(ms) Sleep(ms)

/* 使用pid+clock生成唯一的ipc服务地址,append用于补充功能名称等 */
void IpcApiCreateUrl(IpcApiType type, char* buff, int32_t buffSize, const char* append)
{
    char tmpPath[1024] = {0};
    GetTempPath(sizeof(tmpPath), tmpPath);
    snprintf(buff, buffSize, "ipc://%sipc-%s-%d-%d-%s",
        tmpPath, type == IPCAPI_TYPE_PUBSUB ? "pubsub" : "reqrep", GetCurrentProcessId(), clock(), append);
}

/* ----- windows end ----- */
#else
/* ----- linux begin ----- */
#include <unistd.h>
#include <pthread.h>

#define IPCAPI_SLEEP(ms) usleep(ms * 1000)

/* 使用pid+clock生成唯一的ipc服务地址,append用于补充功能名称等 */
void IpcApiCreateUrl(IpcApiType type, char* buff, int32_t buffSize, const char* append)
{
    snprintf(buff, buffSize, "ipc:///tmp/ipc-%s-%d-%ld-%s", type == IPCAPI_TYPE_PUBSUB ? "pubsub" : "reqrep", getpid(), clock(), append);
}
/* ----- linux end ----- */
#endif

static void IpcApiDeleteUrl(const char* url)
{
	if (strstr(url, "ipc://") == url) {
		remove(url + 6);
	}
}

/* -------------------- zmq,nanomsg,nng三种通信方案 -------------------- */

#ifdef IPCAPI_USE_ZMQ
/* ----- IPCAPI_USE_ZMQ begin ----- */
#include "zmq.h"

typedef struct {
	void* context;
	void* socket;
	int32_t isClient;
	IpcApiType type;
	char* addrBackup;
} IpcApiZmqStruct;

static void _IpcApiStructRelease(IpcApiZmqStruct* iazs)
{
	if (iazs->socket) {
		zmq_close(iazs->socket);
	}
	if (iazs->context) {
		zmq_ctx_shutdown(iazs->context);
		zmq_ctx_term(iazs->context);
	}
	if (iazs->addrBackup) {
		if (!iazs->isClient) {
			IpcApiDeleteUrl(iazs->addrBackup);
		}
		free(iazs->addrBackup);
	}
}
static void _IpcApiRelease(void* handle)
{
	IpcApiZmqStruct* iazs = (IpcApiZmqStruct*)handle;
	if (iazs)
	{
		_IpcApiStructRelease(iazs);
		free(iazs);
	}
}
static void* _IpcApiServer(const char* addr, IpcApiType type)
{
	IpcApiZmqStruct* iazs;
	int32_t zmqType;
	int32_t linger = 1; /* zmq_close()时如果发送缓冲区有数据,滞留时间 */

	int major_ = 0, minor_ = 0, patch_ = 0;
	zmq_version(&major_, &minor_, &patch_);
	IPCAPI_INFO("zmq version: %d.%d.%d \r\n", major_, minor_, patch_);

	if (!addr) {
		return NULL;
	}

	if (type == IPCAPI_TYPE_REQREP) {
		zmqType = ZMQ_REP;
	}
	else if (type == IPCAPI_TYPE_PUBSUB) {
		zmqType = ZMQ_PUB;
	}
	else {
		return NULL;
	}

	iazs = (IpcApiZmqStruct*)calloc(1, sizeof(IpcApiZmqStruct));
	iazs->context = zmq_ctx_new();
	iazs->socket = zmq_socket(iazs->context, zmqType);
	zmq_setsockopt(iazs->socket, ZMQ_LINGER, &linger, sizeof(linger));

	iazs->addrBackup = (char*)calloc(1, strlen(addr) + 1);
	strcpy(iazs->addrBackup, addr);
	iazs->type = type;

	if (zmq_bind(iazs->socket, addr) != 0)
	{
		_IpcApiRelease(iazs);
		IPCAPI_ERR("zmq_bind %s failed \r\n", addr);
		return NULL;
	}
	return iazs;
}
static void* _IpcApiClient(const char* addr, IpcApiType type)
{
	IpcApiZmqStruct* iazs;
	int32_t zmqType;
	int32_t linger = 0; /* zmq_close()时如果发送缓冲区有数据,滞留时间 */

	if (!addr) {
		return NULL;
	}

	if (type == IPCAPI_TYPE_REQREP) {
		zmqType = ZMQ_REQ;
	}
	else if (type == IPCAPI_TYPE_PUBSUB) {
		zmqType = ZMQ_SUB;
	}
	else {
		return NULL;
	}

	iazs = (IpcApiZmqStruct*)calloc(1, sizeof(IpcApiZmqStruct));
	iazs->context = zmq_ctx_new();
	iazs->socket = zmq_socket(iazs->context, zmqType);
	zmq_setsockopt(iazs->socket, ZMQ_LINGER, &linger, sizeof(linger));

	iazs->addrBackup = (char*)calloc(1, strlen(addr) + 1);
	strcpy(iazs->addrBackup, addr);
	iazs->type = type;
	iazs->isClient = 1;

	if (zmq_connect(iazs->socket, addr) != 0)
	{
		_IpcApiRelease(iazs);
		IPCAPI_ERR("zmq_connect %s failed \r\n", addr);
		return NULL;
	}
	return iazs;
}
static int32_t _IpcApiSend(void* handle, const void* buff, int32_t buffLen, int32_t timeout)
{
	IpcApiZmqStruct* iazs = (IpcApiZmqStruct*)handle;
	IpcApiZmqStruct* iazsBackup;
	int32_t ret;
	zmq_setsockopt(iazs->socket, ZMQ_SNDTIMEO, &timeout, sizeof(timeout));
	ret = zmq_send(iazs->socket, buff, buffLen, 0);
	/* 发送失败的客户端,主动重连 */
	if (ret < 0 && iazs->isClient)
	{
		iazsBackup = (IpcApiZmqStruct*)_IpcApiClient(iazs->addrBackup, iazs->type);
		/* 移花接木 */
		if (iazsBackup)
		{
			IPCAPI_INFO("reconnect %s \r\n", iazsBackup->addrBackup);
			_IpcApiStructRelease(iazs);
			memcpy(iazs, iazsBackup, sizeof(IpcApiZmqStruct));
			free(iazsBackup);
			/* 再接再厉 */
			ret = zmq_send(iazs->socket, buff, buffLen, 0);
		}
		else {
			IPCAPI_ERR("create connect %s failed \r\n", iazs->addrBackup);
		}
	}
	return ret;
}
static int32_t _IpcApiRecv(void* handle, void* buff, int32_t buffLen, int32_t timeout)
{
	IpcApiZmqStruct* iazs = (IpcApiZmqStruct*)handle;
	zmq_setsockopt(iazs->socket, ZMQ_RCVTIMEO, &timeout, sizeof(timeout));
	return zmq_recv(iazs->socket, buff, buffLen, timeout < 1 ? ZMQ_DONTWAIT : 0);
}
static int32_t _IpcApiSetTopic(void* handle, const char* topic)
{
	IpcApiZmqStruct* iazs = (IpcApiZmqStruct*)handle;
	return zmq_setsockopt(iazs->socket, ZMQ_SUBSCRIBE, topic, strlen(topic));
}
/* ----- IPCAPI_USE_ZMQ end ----- */
#elif defined(IPCAPI_USE_NANOMSG)
/* ----- IPCAPI_USE_NANOMSG begin ----- */
#include <nanomsg/bus.h>
#include <nanomsg/nn.h>
#include <nanomsg/pair.h>
#include <nanomsg/pipeline.h>
#include <nanomsg/pubsub.h>
#include <nanomsg/reqrep.h>
#include <nanomsg/survey.h>

typedef struct {
	int32_t fd;
	int32_t eid;
	int32_t isClient;
	IpcApiType type;
	char* addrBackup;
} IpcApiNanomsgStruct;
static void _IpcApiRelease(void* _ctx)
{
	IpcApiNanomsgStruct* ctx = (IpcApiNanomsgStruct*)_ctx;
	if (!ctx) {
		return;
	}
	if (ctx->eid >= 0) {
		if (ctx->type == IPCAPI_TYPE_PUBSUB && !ctx->isClient) {
			nn_shutdown(ctx->eid, ctx->fd);
		}
		ctx->eid = -1;
	}
	if (ctx->fd >= 0) {
		nn_close(ctx->fd);
		ctx->fd = -1;
	}
	if (ctx->addrBackup) {
		if (!ctx->isClient) {
			IpcApiDeleteUrl(ctx->addrBackup);
		}
		free(ctx->addrBackup);
	}
	free(ctx);
}
static void* _IpcApiServer(const char* addr, IpcApiType type)
{
	int32_t fd = -1;
	int32_t eid = -1;
	IpcApiNanomsgStruct* ret = NULL;

	/* REQREP 服务端 */
	if (type == IPCAPI_TYPE_REQREP)
	{
		fd = nn_socket(AF_SP, NN_REP);
		if (fd >= 0)
		{
			eid = nn_bind(fd, addr);
			if (eid < 0)
			{
				IPCAPI_ERR("nn_bind failed \"%s\"\r\n", nn_strerror(nn_errno()));
				nn_close(fd);
				return NULL;
			}
		}
	}
	/* PUBSUB 服务端 */
	else if (type == IPCAPI_TYPE_PUBSUB)
	{
		fd = nn_socket(AF_SP, NN_PUB);
		if (fd >= 0)
		{
			eid = nn_bind(fd, addr);
			if (eid < 0)
			{
				IPCAPI_ERR("nn_bind failed \"%s\"\r\n", nn_strerror(nn_errno()));
				nn_close(fd);
				return NULL;
			}
		}
	}
	else {
		return NULL;
	}

	ret = (IpcApiNanomsgStruct*)calloc(1, sizeof(IpcApiNanomsgStruct));
	ret->fd = fd;
	ret->eid = eid;
	ret->type = type;
	ret->addrBackup = (char*)calloc(1, strlen(addr) + 1);
	strcpy(ret->addrBackup, addr);
	return (void*)ret;
}
static void* _IpcApiClient(const char* addr, IpcApiType type)
{
	int32_t fd = -1;
	int32_t eid = -1;
	IpcApiNanomsgStruct* ret = NULL;

	/* REQREP 客户端 */
	if (type == IPCAPI_TYPE_REQREP)
	{
		fd = nn_socket(AF_SP, NN_REQ);
		if (fd >= 0)
		{
			eid = nn_connect(fd, addr);
			if (eid < 0)
			{
				IPCAPI_ERR("nn_connect failed \"%s\"\r\n", nn_strerror(nn_errno()));
				nn_close(fd);
				return NULL;
			}
		}
	}
	/* PUBSUB 客户端 */
	else if (type == IPCAPI_TYPE_PUBSUB)
	{
		fd = nn_socket(AF_SP, NN_SUB);
		if (fd >= 0)
		{
			eid = nn_connect(fd, addr);
			if (eid < 0)
			{
				IPCAPI_ERR("nn_connect failed \"%s\"\r\n", nn_strerror(nn_errno()));
				nn_close(fd);
				return NULL;
			}
		}
	}
	else {
		return NULL;
	}

	ret = (IpcApiNanomsgStruct*)calloc(1, sizeof(IpcApiNanomsgStruct));
	ret->fd = fd;
	ret->eid = eid;
	ret->type = type;
	ret->isClient = 1;
	ret->addrBackup = (char*)calloc(1, strlen(addr) + 1);
	strcpy(ret->addrBackup, addr);
	return (void*)ret;
}
static int32_t _IpcApiSend(void* handle, const void* buff, int32_t buffLen, int32_t timeout)
{
	if (timeout > 0) {
		nn_setsockopt(*((int32_t*)handle), 0, NN_SNDTIMEO, &timeout, sizeof(timeout));
	}
	return nn_send(*((int32_t*)handle), buff, buffLen, 0); /* 0/非阻塞 1/阻塞 */
}
static int32_t _IpcApiRecv(void* handle, void* buff, int32_t buffLen, int32_t timeout)
{
	if (timeout > 0) {
		nn_setsockopt(*((int32_t*)handle), 0, NN_RCVTIMEO, &timeout, sizeof(timeout));
	}
	return nn_recv(*((int32_t*)handle), buff, buffLen, 0); /* 0/非阻塞 1/阻塞 */
}
static int32_t _IpcApiSetTopic(void* handle, const char* topic)
{
	return nn_setsockopt(*((int32_t*)handle), NN_SUB, NN_SUB_SUBSCRIBE, topic, strlen(topic));
}

/* ----- IPCAPI_USE_NANOMSG end ----- */
#endif

/* -------------------- 通用代码 -------------------- */

typedef struct {
	void* handle;
	volatile int32_t threadStateCtrl; /* 0/无 1/正在启动 2/运行中 3/请求退出 */
	void* priv;
	IpcApiCallback callback;
} IpcApiStruct;

/* 抛线程工具 */
#if defined(WIN32) || defined(WIN64) || defined(WINCE) || defined(_MSC_VER)
/* ----- windows begin ----- */
static void _NewThread(void* obj, void (*callback)(void*))
{
	_beginthread(callback, 0, obj);
}
/* 接收和回调线程 */
static void _RecvThread(void* argv)
{
	IpcApiStruct* ips = (IpcApiStruct*)argv;
	int32_t ret;

	char *buff = (char*)calloc(1, IPCAPI_BUFF_SIZE);
	if (!buff)
	{
		IPCAPI_ERR("ipc _RecvThread calloc buff size/%d failed \r\n", IPCAPI_BUFF_SIZE);
		ips->threadStateCtrl = 0;
		return;
	}

	ips->threadStateCtrl = 2;
	while (ips->threadStateCtrl == 2)
	{
		memset(buff, 0, IPCAPI_BUFF_SIZE);
		ret = _IpcApiRecv(ips->handle, buff, IPCAPI_BUFF_SIZE, 100);
		if (ret > 0 && ips->callback) {
			ips->callback(ips, ips->priv, buff, ret);
		}
		IPCAPI_SLEEP(1);
	}

	free(buff);
	ips->threadStateCtrl = 0;
	return;
}
/* ----- windows end ----- */
#else
/* ----- linux begin ----- */
static void _NewThread(void* obj, void* (*callback)(void*))
{
	pthread_t th;
	pthread_attr_t attr;
	/* attr init */
	pthread_attr_init(&attr);
	/* 禁用线程同步, 线程运行结束后自动释放 */
	pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);
	/* 抛出线程 */
	pthread_create(&th, &attr, callback, (void*)obj);
	/* attr destroy */
	pthread_attr_destroy(&attr);
}
/* 接收和回调线程 */
static void* _RecvThread(void* argv)
{
	IpcApiStruct* ips = (IpcApiStruct*)argv;
	int32_t ret;

	char *buff = (char*)calloc(1, IPCAPI_BUFF_SIZE);
	if (!buff)
	{
		ips->threadStateCtrl = 0;
		return NULL;
	}

	ips->threadStateCtrl = 2;
	while (ips->threadStateCtrl == 2)
	{
		memset(buff, 0, IPCAPI_BUFF_SIZE);
		ret = _IpcApiRecv(ips->handle, buff, IPCAPI_BUFF_SIZE, 100);
		if (ret > 0 && ips->callback) {
			ips->callback(ips, ips->priv, buff, ret);
		}
		IPCAPI_SLEEP(1);
	}

	free(buff);
	ips->threadStateCtrl = 0;
	return NULL;
}
/* ----- linux endif ----- */
#endif

static void* _IpcApiServerOrClient(const char* addr, IpcApiType type, void* priv, IpcApiCallback callback, int32_t isServer)
{
	IpcApiStruct* ret = NULL;
	void* handle = isServer ? _IpcApiServer(addr, type) : _IpcApiClient(addr, type);
	if (handle)
	{
		ret = (IpcApiStruct*)calloc(1, sizeof(IpcApiStruct));
		ret->handle = handle;
		ret->priv = priv;
		ret->callback = callback;
		if (ret->callback)
		{
			ret->threadStateCtrl = 1;
			_NewThread(ret, _RecvThread);
		}
	}
	return ret;
}

void* IpcApiServer(const char* addr, IpcApiType type, void* priv, IpcApiCallback callback)
{
	return _IpcApiServerOrClient(addr, type, priv, callback, 1);
}

void* IpcApiClient(const char* addr, IpcApiType type, void* priv, IpcApiCallback callback)
{
	return _IpcApiServerOrClient(addr, type, priv, callback, 0);
}

void IpcApiRelease(void* handle)
{
	IpcApiStruct* ips = (IpcApiStruct*)handle;
	int32_t timeoutMs = 500;
	if (ips)
	{
		if (ips->threadStateCtrl == 2) {
			ips->threadStateCtrl = 3;
			while (ips->threadStateCtrl && timeoutMs > 0) {
				IPCAPI_SLEEP(5);
				timeoutMs -= 5;
			}
		}
		if (ips->handle) {
			_IpcApiRelease(ips->handle);
		}
		free(ips);
	}
}

int32_t IpcApiSend(void* handle, const void* buff, int32_t buffLen, int32_t timeout)
{
	IpcApiStruct* ips = (IpcApiStruct*)handle;
	if (!ips || !ips->handle || !buff || buffLen < 1) {
		return -1;
	}
	return _IpcApiSend(ips->handle, buff, buffLen, timeout);
}

int32_t IpcApiRecv(void* handle, void* buff, int32_t buffLen, int32_t timeout)
{
	IpcApiStruct* ips = (IpcApiStruct*)handle;
	if (!ips || !ips->handle || !buff || buffLen < 1) {
		return -1;
	}
	return _IpcApiRecv(ips->handle, buff, buffLen, timeout);
}

int32_t IpcApiSetTopic(void* handle, const char* topic)
{
	IpcApiStruct* ips = (IpcApiStruct*)handle;
	if (!ips || !ips->handle) {
		return -1;
	}
	return _IpcApiSetTopic(ips->handle, topic);
}
